package com.service.websocket.handler;


import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.service.websocket.service.WebSocketTaskHandler;
import com.service.websocket.to.User;
import com.service.websocket.to.MessageTO;
import com.service.websocket.to.ResponseObject;
import com.service.websocket.to.Topic;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author xianggj
 * @description: 消息处理中心
 * @date 2019/8/20 16:29
 */
@Slf4j
@Component
public class MessageHandler extends TextWebSocketHandler {

    /**
     * 在线用户列表
     */
    private static final Map<String, Map<String, WebSocketSession>> USER_LIST;

    /**
     * 在线业务列表
     */
    private static final Map<String, Map<String, WebSocketSession>> BUS_LIST;

    static {
        USER_LIST = new ConcurrentHashMap<>();
        BUS_LIST = new ConcurrentHashMap<>();
    }

    /**
     * 用户标识
     */
    public static final String USER_ID = "userId";
    /**
     * 用户标识
     */
    public static final String HEADER = "header";

    /**
     * 业务标识
     */
    public static final String BUS_ID = "busId";

    /**
     * 用户登录信息
     */
    public static final String TOKEN = "token";

    /**
     * 用户标识
     */
    public static final String USER_NAME = "userName";

    /**
     * 窗口标识
     */
    public static final String WINDOW_ID = "winId";

    @Autowired
    protected WebSocketTaskHandler webSocketTaskHandler;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("成功建立连接");
        String userId = getUserId(session);
        String busId= getBusId(session);
        if (StrUtil.isBlank(userId)){
            session.sendMessage(new TextMessage("未获取到用户信息，链接自动断开！"));
            session.close();
            return;
        }
        String windowId = getWindowId(session);
        String userName = getUserName(session);
        StringBuffer sb = new StringBuffer("连接成功，请等待消息...");
        if (StrUtil.isNotBlank(userId)) {
            putUserList(session, userId);
            log.info("clientId : " + userId);
            log.info("userName : " + userName);
        }
        if (StrUtil.isNotBlank(busId)){
            putBusList(session, busId);
            log.info("busId : " + busId);
        }
        sendWelcomeMessage(session, windowId, sb.toString());
    }

    /**
     * 发送欢迎信息
     * @param session
     * @param windowId
     * @throws IOException
     */
    private void sendWelcomeMessage(WebSocketSession session, String windowId, String welcomeStr) throws IOException {
        ResponseObject responseObject = new ResponseObject();
        MessageTO messageTO = new MessageTO(Topic.WIN_ID.getValue(),
                welcomeStr, windowId);
        responseObject.setStatus(HttpStatus.OK.value());
        responseObject.setMessage("成功建立websocket链接");
        responseObject.setData(messageTO);
        String s = JSON.toJSONString(responseObject, SerializerFeature.WriteMapNullValue) + "";
        session.sendMessage(new TextMessage(s));
    }


    /**
     *  添加用户session
     * @param session
     * @param clientId
     */
    protected void putUserList(WebSocketSession session, String clientId) {
        String winId = getWindowId(session);
        if (USER_LIST.containsKey(clientId)){
            USER_LIST.get(clientId).put(winId,session);
        } else {
            Map<String, WebSocketSession> map = new HashMap<>();
            map.put(winId, session);
            USER_LIST.put(clientId, map);
        }
    }

    /**
     *  添加业务session
     * @param session
     * @param busId
     */
    protected void putBusList(WebSocketSession session, String busId) {
        String winId = getWindowId(session);
        if (busId != null && BUS_LIST.containsKey(busId)){
            BUS_LIST.get(busId).put(winId,session);
        } else {
            Map<String, WebSocketSession> map = new HashMap<>();
            map.put(winId, session);
            BUS_LIST.put(busId, map);
        }
    }

    /**
     * 删除链接
     * @param session
     */
    protected void remove(WebSocketSession session) {
        String winId = getWindowId(session);
        String userId = getUserId(session);
        if (USER_LIST != null && StrUtil.isNotEmpty(userId) && USER_LIST.containsKey(userId)){
            Map<String, WebSocketSession> map = USER_LIST.get(userId);
            map.remove(winId);
            //当用户最后一个窗口退出时 进行删除
            if (map.size() == 0){
                USER_LIST.remove(userId);
            }
        }
        String busId = getBusId(session);
        if (busId != null && BUS_LIST.containsKey(busId)){
            Map<String, WebSocketSession> map = BUS_LIST.get(busId);
            map.remove(winId);
            if (map.size() == 0){
                BUS_LIST.remove(busId);
            }
        }
        //关闭连接时 退出所有的队列
        webSocketTaskHandler.removeKeyByWinId(winId);
    }

    /**
     * 重复请求间隔
     * 默认100毫秒
     * 当同一个窗口0.1秒内重复发送一个请求时，任务这个是重复请求，不予处理
     */
    private static final Long INTERVAL = 100L;

    /**
     * 公共缓存key
     */
    private static final WeakHashMap<String, Long> weakHashMap = new WeakHashMap<>();


    /**
     * 处理消息
     * @param session
     * @param message
     */
    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        if (message == null){
            return;
        }
        try {
            String str = message.getPayload();
            log.info("websocket接收到消息："+ str);
            if (StrUtil.isNotBlank(str)){
                try {
                    MessageTO messageTO = JSON.parseObject(str, MessageTO.class);
                    String windowId = getWindowId(session);
                    String userId = getUserId(session);
                    messageTO.setWindowId(windowId);
                    messageTO.setUserId(userId);
                    //心跳检测
                    if (Topic.HEART_BEAT.getValue().equalsIgnoreCase(messageTO.getTopic())){
                        messageTO.setData("链接刷新成功");
                        //封装统一格式
                        ResponseObject responseObject = new ResponseObject();
                        responseObject.setStatus(HttpStatus.OK.value());
                        responseObject.setMessage(ResponseObject.SUCCESS_MESSAGE);
                        responseObject.setData(messageTO);
                        sendMessageToWindow(userId, responseObject, windowId);
                        return;
                    }
                    String token = getToken(session);
                    messageTO.setToken(token);

                    String header = getHeader(session);
                    String userName = getUserName(session);
                    messageTO.setHeader(header);
                    messageTO.setUserName(userName);
                    String busId = getBusId(session);
                    messageTO.setBusId(busId);
                    long startTime = System.currentTimeMillis();
                    messageTO.setTime(startTime);
                    webSocketTaskHandler.addPushQueue(messageTO);
                } catch (Exception e) {
                    log.info("无效消息:" + str);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送信息给指定用户
     * @param userId
     * @param messageTO 消息对象
     * @return
     */
    public boolean sendMessageToUser(String userId, Object messageTO) {
        TextMessage message = new TextMessage(JSON.toJSONString(messageTO,
                SerializerFeature.WriteMapNullValue) + "");
        if (USER_LIST.get(userId) == null) {
            return false;
        }
        boolean status = true;
        Map<String, WebSocketSession> sessions = USER_LIST.get(userId);
        for (Map.Entry<String, WebSocketSession> entry: sessions.entrySet()) {
            WebSocketSession session = entry.getValue();
            if (!session.isOpen()) {
                continue;
            }
            try {
                //防止多用户同时使用这个session
                synchronized (session) {
                    session.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
                status = false;
            }
        }
        return status;
    }

    /**
     * 发送信息给当前窗口
     * @param userId
     * @param messageTO 消息对象
     * @return
     */
    public boolean sendMessageToWindow(String userId, Object messageTO, String winId) {
        TextMessage message = new TextMessage(JSON.toJSONString(messageTO)+"");
        if (!isExits(userId, winId)) {
            return false;
        }
        Map<String, WebSocketSession> sessions = USER_LIST.get(userId);
        WebSocketSession session = sessions.get(winId);
        try {
            //防止多用户同时使用这个session
            synchronized (session) {
                session.sendMessage(message);
            }
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 根据业务发送请求
     * @param busId
     * @param messageTO 消息对象
     * @return
     */
    public boolean sendMessageByBusId(String busId, Object messageTO) {
        TextMessage message = new TextMessage(JSON.toJSONString(messageTO)+"");
        if (StrUtil.isBlank(busId) || BUS_LIST.get(busId) == null) {
            return false;
        }
        boolean status = true;
        Map<String, WebSocketSession> sessions = BUS_LIST.get(busId);
        for (Map.Entry<String, WebSocketSession> entry: sessions.entrySet()) {
            WebSocketSession session = entry.getValue();
            if (!session.isOpen()) {
                continue;
            }
            try {
                //防止多用户同时使用这个session
                synchronized (session) {
                    session.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
                status = false;
            }
        }
        return status;
    }

    /**
     * 根据业务发送请求-除了自己窗口
     * @param busId
     * @param messageTO 消息对象
     * @return
     */
    public boolean sendMessageByBusIdExceptMyWin(String busId, Object messageTO, String winId) {
        TextMessage message = new TextMessage(JSON.toJSONString(messageTO)+"");
        if (StrUtil.isBlank(busId) || BUS_LIST.get(busId) == null) {
            return false;
        }
        boolean status = true;
        Map<String, WebSocketSession> sessions = BUS_LIST.get(busId);
        for (Map.Entry<String, WebSocketSession> entry: sessions.entrySet()) {
            //自己窗口 过滤
            if (entry.getKey().equalsIgnoreCase(winId)){
                continue;
            }
            WebSocketSession session = entry.getValue();
            if (!session.isOpen()) {
                status = false;
                continue;
            }
            try {
                //防止多用户同时使用这个session
                synchronized (session) {
                    session.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
                status = false;
            }
        }
        return status;
    }
    /**
     * 根据业务发送请求-除了自己用户
     * @param busId
     * @param messageTO 消息对象
     * @return
     */
    public boolean sendMessageByBusIdExceptOneSelf(String busId, Object messageTO, String userId) {
        TextMessage message = new TextMessage(JSON.toJSONString(messageTO)+"");
        if (StrUtil.isBlank(busId) || BUS_LIST.get(busId) == null) {
            return false;
        }
        boolean status = true;
        Map<String, WebSocketSession> sessions = BUS_LIST.get(busId);
        for (WebSocketSession session: sessions.values()) {
            String sessionUserId = getUserId(session);
            //自己用户 过滤
            if (sessionUserId.equalsIgnoreCase(userId)){
                continue;
            }
            if (!session.isOpen()) {
                status = false;
                continue;
            }
            try {
                //防止多用户同时使用这个session
                synchronized (session) {
                    session.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
                status = false;
            }
        }
        return status;
    }

    /**
     * 广播信息
     * @return
     */
    public Set<String> sendMessageToAllUsers(Object messageTO) {
        TextMessage message = new TextMessage(JSON.toJSONString(messageTO)+"");
        Set<String> set = new HashSet<>();
        Set<String> clientIds = USER_LIST.keySet();
        for (String clientId : clientIds) {
            try {
                Map<String, WebSocketSession> sessions = USER_LIST.get(clientId);
                for (Map.Entry<String, WebSocketSession> entry: sessions.entrySet()) {
                    WebSocketSession session = entry.getValue();
                    if (session.isOpen()) {
                        synchronized (session) {
                            session.sendMessage(message);
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
                set.add(clientId);
            }
        }

        return set;
    }


    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        exception.printStackTrace();
        if (session.isOpen()) {
            session.close();
        }
        log.info("连接出错");
        remove(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String windowId = getWindowId(session);
        log.info("连接已关闭：" + windowId);
        log.info("状态："+status);
        remove(session);
        Collection<AbstractMessageHandler> values = 
                WebSocketTaskHandler.PRECESSOR_MAP.values();
        //关闭窗口后 清除所有的windowId相关操作
        for (AbstractMessageHandler amh : values) {
            if (amh != null){
                amh.afterConnectionClosed(windowId);
            }
        }

    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    /**
     * 获取用户标识
     * @param session
     * @return
     */
    protected String getUserId(WebSocketSession session) {
        try {
            String userId = (String) session.getAttributes().get(USER_ID);
            return userId;
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 获取窗口标识
     * @param session
     * @return
     */
    protected String getWindowId(WebSocketSession session) {
        try {
            String winId = (String) session.getAttributes().get(WINDOW_ID);
            return winId;
        } catch (Exception e) {
            return null;
        }
    }
    /**
     * 获取窗口标识
     * @param session
     * @return
     */
    protected String getToken(WebSocketSession session) {
        try {
            String token = (String) session.getAttributes().get(TOKEN);
            return token;
        } catch (Exception e) {
            return null;
        }
    }

    protected String getUserName(WebSocketSession session) {
        try {
            String userName = (String) session.getAttributes().get(USER_NAME);
            return userName;
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 获取窗口标识
     * @param session
     * @return
     */
    protected String getBusId(WebSocketSession session) {
        try {
            String busId = (String) session.getAttributes().get(BUS_ID);
            return busId;
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 获取请求头
     * @param session
     * @return
     */
    protected String getHeader(WebSocketSession session) {
        try {
            String header = (String) session.getAttributes().get(HEADER);
            return header;
        } catch (Exception e) {
            return null;
        }
    }


    /**
     * 判断session是否存在
     * @return
     */
    protected boolean isExits(String winId){
        return isExits(null, winId);
    }

    /**
     * 判断session是否存在
     * 这里只判断用户id
     *  业务id本就可能不存在
     * @param winId
     * @return
     */
    public boolean isExits(String userId, String winId){
        if (StrUtil.isBlank(userId)){
            return false;
        }
        if (USER_LIST.get(userId) == null) {
            return false;
        }
        Map<String, WebSocketSession> sessions = USER_LIST.get(userId);
        if (sessions.size() == 0){
            return false;
        }
        //如果存在
        if (StrUtil.isNotBlank(winId)){
            if (sessions.containsKey(winId)){
                WebSocketSession session = sessions.get(winId);
                if (!session.isOpen()){
                    return false;
                }
            } else {
                return false;
            }
        }
        return true;
    }


    /**
     * 判断session是否存在
     * 这里只判断用户id
     *  业务id本就可能不存在
     * @param busId
     * @return
     */
    public User isExitsBusId(String busId){
        if (StrUtil.isBlank(busId)){
            return null;
        }
        if (BUS_LIST.get(busId) == null) {
            return null;
        }
        Map<String, WebSocketSession> sessions = BUS_LIST.get(busId);
        if (sessions.size() == 0){
            return null;
        }
        for (WebSocketSession session : sessions.values()) {
            if (session.isOpen()){
                return new User(getUserId(session), getWindowId(session), getUserName(session));
            }
        }
        return null;
    }


}
